package com.hm.rabitmq.distributing;

import java.util.Random;
import java.util.concurrent.TimeUnit;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

/**
 * @author shishun.wang
 * @date 2016年12月9日 上午11:16:52
 * @version 1.0
 * @describe 消费者，处理任务
 */
public class Worker {

	private final static String QUEUE_NAME = "WORK_QUEUE";

	private final static Random random = new Random();

	public static void main(String[] args) {
		try {
			accept();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void accept() throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("115.28.66.183");

		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		boolean durable = true; // 声明队列可持久化(消费者和生产者均需要)，解决由于消费者突然挂机消息丢失问题
		channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
		 //公平分发
        channel.basicQos(1);

		QueueingConsumer consumer = new QueueingConsumer(channel);
		boolean autoAck = false;// 开启消息应答机制
		{
			// 在上面的例子中，如果某个worker(比如worker
			// 1)在处理任务过程中突然死掉，那么task1，task4,task7三个任务就会丢失。可以通过消息应答机制来解决这个问题。开启消息应答后，
			// 如果消费者接收并处理完消息后会向rabbitmq发送消息应答，rabbitmq收到消息应答后才会把对应的消息删除；如果消费者由于死掉导致没有发送消息应答，
			// rabbitmq会认为消息没有被完全处理，从而把对应消息分发给下一个消费者处理。
			// 默认情况下，rabbitmq是启用消息应答的,即autoAck=true,
		}
		channel.basicConsume(QUEUE_NAME, autoAck, consumer);

		System.out.println("waiting for task....");
		while (true) {
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String task = new String(delivery.getBody());
			System.out.println("worker start to handler the task:" + task);
			// 模拟处理任务花费时间
			handlerTask();
			System.out.println("worker end to handler the task:" + task);
			
			//消息应答,开启消息接收到反馈
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
		}
	}

	private static void handlerTask() {
		try {
			TimeUnit.SECONDS.sleep(random.nextInt(5) + 1);
		} catch (InterruptedException e) {
			// ignore
		}
	}
}
